package com.whale.springboot.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;

@Configuration
@Slf4j
public class KafkaConfig {

    /**
     * 消费异常处理器
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
                //打印消费异常的消息和异常信息
                log.error("consumer failed! message: {}, exceptionMsg: {}, groupId: {}", message, exception.getMessage(), exception.getGroupId());
                return null;
            }
        };
    }

    /**
     * 通过注入一个 NewTopic 类型的 Bean 来创建 topic，如果 topic 已存在，则会忽略
     */
    @Bean
    public NewTopic myTopic() {
        return new NewTopic("hello-kafka-test-topic", 3, (short) 3);
    }

    /**
     * 通过注入一个 NewTopic 类型的 Bean 来创建 topic，如果 topic 已存在，则会忽略
     */
    @Bean
    public NewTopic myTopic2() {
        return new NewTopic("hello-kafka-test-topic-exception", 3, (short) 3);
    }
    @Bean
    public NewTopic myTopic3() {
        return new NewTopic("consumer-interceptor", 3, (short) 3);
    }
    @Bean
    public NewTopic myTopic4() {
        return new NewTopic("hello-batch", 3, (short) 3);
    }
    @Bean
    public NewTopic myTopic5() {
        return new NewTopic("jkhtest6", 3, (short) 3);
    }

}
